package fox.framework.rpc.client.register;

import fox.framework.rpc.client.AbstractPushRpcRegister;
import fox.framework.rpc.client.RegisterNodeInfo;
import fox.framework.rpc.client.node.RpcNode;
import fox.framework.rpc.common.ExceptionUtils;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.data.Stat;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author cuichao
 * @Description: ZookeeperRegister
 * @Date: create in 2021/1/10 22:54
 */
public class ZookeeperRegister extends AbstractPushRpcRegister {


    public static final String RPC_ROOT_PATH = "fox";
    public static final String RPC_NODE_LIST_PATH = "nodes";

    private String zkServers = "127.0.0.1:2181";
    private int maxRetires = 3;
    private int retryIntervalMs = 3000;
    private int sessionTimeoutMs = 30000;
    private int connectionTimeoutMs = 300000;
    private ZkClient zkClient;


    public ZookeeperRegister() {
        super();
    }


    public ZookeeperRegister(String zkServers) {
        this.zkServers = zkServers;
    }

    public ZookeeperRegister(String zkServers, int maxRetires, int retryIntervalMs, int sessionTimeoutMs, int connectionTimeoutMs) {
        this.zkServers = zkServers;
        this.maxRetires = maxRetires;
        this.retryIntervalMs = retryIntervalMs;
        this.sessionTimeoutMs = sessionTimeoutMs;
        this.connectionTimeoutMs = connectionTimeoutMs;
    }

    @Override
    protected List<RegisterNodeInfo> fetchNodeData(List<String> metaspaces) {
        try {
            List<RegisterNodeInfo> infoList = zkClient.fetchData(metaspaces);
            zkClient.metaspaceWatcher(metaspaces, this);
            return infoList;
        } catch (Exception e) {
            throw ExceptionUtils.RpcRuntimeExceptionWrap(e);
        }

    }

    @Override
    public void register(String address, List<String> metaspaces) throws Exception {
        zkClient.register(address, metaspaces);
    }

    @Override
    public void run() throws Exception {
        zkClient = new ZkClient();
        zkClient.start();
    }

    @Override
    public void close() throws Exception {
        zkClient.close();
    }


    private class ZkClient {

        private CuratorFramework client;
        private Map<String, PathChildrenCache> cacheMap = new HashMap<>(32);
        private ExecutorService WatcherPool = new ThreadPoolExecutor(
                16, 32, 60L,
                TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1024),
                v -> new Thread(v, "zookeeper-cache-thread"));


        private List<RegisterNodeInfo> fetchData(List<String> metaspaces) throws Exception {
            List<RegisterNodeInfo> infoList = new ArrayList<>();
            if(metaspaces == null || metaspaces.isEmpty()){
                return infoList;
            }
            for (String metaspace : metaspaces) {
                //判断zk是否有此节点 没有则创建节点
                String path = buildNodePath(metaspace);
                Stat stat = client.checkExists().forPath(path);
                if(stat == null){
                    client.create().creatingParentsIfNeeded().forPath(path);
                }
                //查看其子节点得到远程服务节点数据
                List<String> children = client.getChildren().forPath(path);
                for (String address : children) {
                    infoList.add(new RegisterNodeInfo(metaspace,address,RpcNode.RPC_NODE_READY));
                }
            }
            return infoList;
        }


        private void register(String address, List<String> metaspaces) throws Exception {
            for (String metaspace : metaspaces) {
                String path = buildNodePath(metaspace);
                Stat stat = client.checkExists().forPath(path);
                if (stat == null) {
                    client.create().creatingParentsIfNeeded().forPath(path);
                }
                String nodePath = path+"/"+address;

                if(client.checkExists().forPath(nodePath) != null){
                    client.delete().forPath(nodePath);
                }
                client.create().withMode(CreateMode.EPHEMERAL).forPath(nodePath);

            }
        }


        private void metaspaceWatcher(List<String> metaspaces, ZookeeperRegister register) throws Exception {
            for (String metaspace : metaspaces) {
                String path = buildNodePath(metaspace);
                PathChildrenCache cache = new PathChildrenCache(client, path, true);
                cache.getListenable().addListener(addListener(path, register), WatcherPool);
                cacheMap.put(metaspace, cache);
                cache.start();
            }

        }

        private PathChildrenCacheListener addListener(String path, ZookeeperRegister register) {
            return new PathChildrenCacheListener() {
                @Override
                public void childEvent(CuratorFramework curatorFramework, PathChildrenCacheEvent event) throws Exception {
                    switch (event.getType()) {
                        case CHILD_ADDED:
                            //新增或重新上线
                            register.triggerUpdateNodeInfo(path, event.getData().getPath(), RpcNode.RPC_NODE_READY);
                            break;
                        case CHILD_REMOVED:
                            //下线
                            register.triggerUpdateNodeInfo(path, event.getData().getPath(), RpcNode.RPC_NODE_SHUTDOWN);
                            break;
                        default:
                            //nothing to do
                    }
                }
            };
        }

        private void start() {
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(retryIntervalMs, maxRetires);
            this.client = CuratorFrameworkFactory.newClient(zkServers, sessionTimeoutMs, connectionTimeoutMs, retryPolicy);
            client.start();
        }


        private void close() throws Exception {
            try {
                for (Map.Entry<String, PathChildrenCache> entry : cacheMap.entrySet()) {
                    entry.getValue().close();
                }
            } finally {
                client.close();
            }
        }


        private String buildNodePath(String metaspace) {
            return String.format("/%s/%s/%s", RPC_ROOT_PATH, RPC_NODE_LIST_PATH, metaspace);
        }


    }


}
